package com.mjf.day2

import org.apache.flink.streaming.api.scala._

/**
 * 多流转换算子
 *  split-select, connect-comap/coflatmap 成对出现
 *  先转换成 SplitStream，ConnectedStreams，然后再通过select/comap操作转换回来DataStream
 *  所谓coMap，其实就是基于ConnectedStreams的map方法，里面传入的参数是CoMapFunction
 */
object SplitStreamExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream: DataStream[SensorReading] = env.addSource(new SensorSource)

    // 分流，给数据流打标签
    val splitStream: SplitStream[SensorReading] = stream.split(
      data => {
        if (data.temperature > 10) {
          Seq("high")
        } else {
          Seq("low")
        }
      }
    )

    // 分流，选择具体的流
    val highTempStream: DataStream[SensorReading] = splitStream.select("high")
    val lowTempStream: DataStream[SensorReading] = splitStream.select("low")
    val allTempStream: DataStream[SensorReading] = splitStream.select("high", "low")

    highTempStream.print("high")
    lowTempStream.print("low")
    allTempStream.print("all")

    val warningStream: DataStream[(String, Double)] = highTempStream.map(
      data => (data.id, data.temperature)
    )

    // 合流
    val connectedStreams: ConnectedStreams[(String, Double), SensorReading] = warningStream.connect(lowTempStream)

    val result: DataStream[Product] = connectedStreams.map(
      warningData => (warningData._1, warningData._2, "high temp warning"),
      lowTempData => (lowTempData.id, "normal")
    )

    result.print("result")

    env.execute("SplitStreamExample")

  }
}
